package gspushmq.v1.subscribe;

import java.util.Date;
import java.util.concurrent.ConcurrentSkipListSet;
import java.util.concurrent.CopyOnWriteArraySet;

import org.apache.commons.lang3.StringUtils;
import org.java_websocket.WebSocket;

import gspushmq.model.ApiMessage;
import gspushmq.model.ApiMessageType;
import gspushmq.model.DataBase;
import gspushmq.model.Message;
import gspushmq.model.MessageType;
import gspushmq.model.MqType;
import gspushmq.model.User;
import gspushmq.util.ClearMemoryUtil;
import gspushmq.util.DateUtil;
import gspushmq.util.GGLogger;
import gspushmq.util.MqApiComparable;
import gspushmq.util.MqComparable;
import gspushmq.v1.produce.GGDataStorage;
import gspushmq.v1.produce.job.hq.HQService;
import gspushmq.v1.subscribe.core.HandlerSubscribe;
import gspushmq.v1.subscribe.send.core.GGMessageSend;

/**
 * 处理具体的业务逻辑
 *
 * @author yutao
 * @version 1.0
 * @company 上海朝阳永续信息技术有限公司
 * @copyright (c) 2016 SunTime Co'Ltd Inc.All rights reserved.
 * @date 2018年11月1日下午3:06:31
 * @since JDK1.7
 */
public class HandlerSubscribeImpl extends HandlerSubscribe {

    public HandlerSubscribeImpl(Message mess, WebSocket wst) {
        super(mess, wst);
//        this.setPriority(20);
    }

    @Override
    public boolean valid() {
        Message message = getMessage();
        WebSocket wst = getWebSocket();

        boolean flag = true;

        String result = "";
        //连接等待过程中，如果有连接close 则，不就消息回发
        if (!wst.isOpen()) {
            result = "websocket已关闭";
            GGLogger.record(new StringBuffer("action=校验错误").append("|").append(message).append("|result=" + result));
            try {
                throw new Exception("websocket已关闭");
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (message == null) {
            result = "请求参数无效，不允许为空";
        }

        try {
            //订阅消息体的校验
            if (StringUtils.isBlank(result)) {
                message.valid();
            }
        } catch (Exception e) {
            result = e.toString();
        } finally {
            if (StringUtils.isNotBlank(result)) {
                GGLogger.record(new StringBuffer("action=校验错误")
                        .append("|websocket=").append(wst)
                        .append("|ip=").append(wst.getRemoteSocketAddress().getAddress().getHostAddress())
                        .append("|").append(message)
                        .append("|result=" + result));
                wst.send(result);
                flag = false;
            }
        }
        return flag;
    }

    @Override
    public void doHandler() {
        Message message = getMessage();
        WebSocket wst = getWebSocket();
        message.setWst(wst);
        //开始处理具体的业务逻辑
        String mqType = message.getMq_type();
        String mqSubType = message.getMq_sub_type();
        String subStatus = message.getSub_status();

        MessageType messType = new MessageType();
        messType.setMq_sub_type(mqSubType);
        messType.setMq_type(mqType);

        switch (subStatus) {
            case "0":
                switch (mqType) {
                    case "api":

                        cancleApiMessage(message, wst);

                        break;
                    default:
                        break;
                }
                User user = GGSubscribe.wstUser.get(wst);
                GGSubscribe.userWst.remove(user);
                GGSubscribe.wstUser.remove(wst);

                break;
            case "1":
            	ConcurrentSkipListSet<Message> messageSet = GGSubscribe.messWocketMap.get(messType);
                if (messageSet == null) {
                    messageSet = new ConcurrentSkipListSet<Message>(new MqComparable());
                    GGSubscribe.messWocketMap.put(messType, messageSet);
                }
                messageSet.add(message);
                switch (mqType) {
                    case MqType.API:

                        String apiUrl = message.getApi_url();
                        String requestId = message.getRequest_id();
                        String params = message.getParams();

                        ApiMessageType apiType = new ApiMessageType();
                        apiType.setApi_url(apiUrl);
                        apiType.setParams(params);
                        apiType.setReqeust_id(requestId);
                        apiType.setMq_sub_type(message.getMq_sub_type());

                        //api+requestId 和 用户订阅的对于关系
                        ConcurrentSkipListSet<ApiMessage> apiMessSet = GGSubscribe.ApiUrlMessage.get(apiType);
                        if (apiMessSet == null) {
                            apiMessSet = new ConcurrentSkipListSet<ApiMessage>(new MqApiComparable());
                            GGSubscribe.ApiUrlMessage.put(apiType, apiMessSet);
                        }
                        apiMessSet.add((ApiMessage) message);
                        switch (mqSubType) {
                            case MqType.COUNT:
                                break;
                            case MqType.LIST:
                                if ("v1/tg/get_ai_event_catalysts_list".equals(apiUrl)) {
                                    //ai智能推送需要记录订阅时间
                                    DataBase dataBase = GGDataStorage.ApiBaseMap.get(apiType);
                                    if (dataBase == null) {
                                        dataBase = new DataBase();
                                        GGDataStorage.ApiBaseMap.put(apiType, dataBase);
                                    }
                                    //记录订阅时间
                                    dataBase.getBaseDate().put(message.getAccount_id(), DateUtil.dateToString(new Date(), "yyyy-MM-dd HH:mm:ss.SSS"));
                                }

                                break;
                            default:
                                break;
                        }
                        //处理第一次订阅
                        if (!MqType.specialSet.contains(apiUrl)) {
                            HandlerFirstSubscribe.done((ApiMessage) message, wst);
                        }
                        break;
                    case MqType.SYS:

                        switch (mqSubType) {
                            //	互踢
                            case "login":
                                //产品线
                                String product_line = message.getProduct_line();
                                String account_id = message.getAccount_id();
                                User u = new User();
                                u.setAccountId(account_id);
                                u.setProductLine(product_line);
                                WebSocket preWst = GGSubscribe.userWst.get(u);
                                if (preWst == null || preWst.equals(wst)) {
                                    GGSubscribe.userWst.put(u, wst);
                                    return;
                                }
                                GGMessageSend.sendMsgString(message, preWst, "已在另一个地点登录过");
                                ClearMemoryUtil.clearMemory(preWst);
                                GGSubscribe.userWst.put(u, wst);
                                break;
                            default:
                                break;
                        }
                        break;
                    case MqType.HQ:     //行情预警
                        HQService.dealHqWarn(message,wst);
                        break;
                    default:
                        break;
                }
                break;
        }
    }

    /**
     * api取消订阅
     *
     * @param message
     * @param wst
     * @author yutao
     * @date 2018年11月16日下午2:14:01
     */
    private void cancleApiMessage(Message message, WebSocket wst) {
        GGLogger.warn(getClass(), "----取消订阅----");
        ApiMessageType apiType = new ApiMessageType();
        apiType.setApi_url(message.getApi_url());
        apiType.setMq_sub_type(message.getMq_sub_type());
        apiType.setReqeust_id(message.getRequest_id());

        ConcurrentSkipListSet<ApiMessage> apiSet = GGSubscribe.ApiUrlMessage.get(apiType);
        apiSet.remove((ApiMessage) message);
        GGSubscribe.ApiUrlMessage.put(apiType, apiSet);
    }



}
